package com.lry.rpc.v1;


import com.lry.common.ProtocolConstants;
import com.lry.compressor.Compressor;
import com.lry.compressor.CompressorFactory;
import com.lry.protocol.HeartbeatInvocation;
import com.lry.protocol.RpcMessage;
import com.lry.rpc.netty.MsgType;
import com.lry.serialize.HeadMapSerializer;
import com.lry.serialize.Serializer;
import com.lry.serialize.SerializerFactory;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;

/**
 * <pre>
 * 0     1     2     3     4     5     6     7     8     9    10     11    12    13    14    15    16
 * +-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+
 * |   magic   |Proto|     Full length       |    Head   | Msg |Seria|Compr|     RequestId         |
 * |   code    |colVer|    (head+body)      |   Length  |Type |lizer|ess  |                       |
 * +-----------+-----------+-----------+-----------+-----------+-----------+-----------+-----------+
 * |                                                                                               |
 * |                                   Head Map [Optional]                                         |
 * +-----------+-----------+-----------+-----------+-----------+-----------+-----------+-----------+
 * |                                                                                               |
 * |                                         body                                                  |
 * |                                                                                               |
 * |                                        ... ...                                                |
 * +-----------------------------------------------------------------------------------------------+
 * </pre>
 * <p>
 * <li>Full Length: include all data </li>
 * <li>Head Length: include head data from magic code to head map. </li>
 * <li>Body Length: Full Length - Head Length</li>
 * </p>
 */
@Slf4j
public class ProtocolV1Decoder extends LengthFieldBasedFrameDecoder {
    public ProtocolV1Decoder(){
        this(ProtocolConstants.MAX_FRAME_LENGTH);
    }
    public ProtocolV1Decoder(int maxFrameLength) {
        super(maxFrameLength, 3, 4,-7,0);
    }

    @Override
    protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        Object decoded = super.decode(ctx, in);
        if(decoded instanceof ByteBuf){
            ByteBuf frame = (ByteBuf) decoded;
            try {
                return decodeFrame(frame);
            } catch (Exception e) {
                log.error("Decode frame error!", e);
                throw e;
            } finally {
                frame.release();
            }
        }
        return decoded;
    }

    private Object decodeFrame(ByteBuf frame) {
        byte b0 = frame.readByte();
        byte b1 = frame.readByte();
        if (ProtocolConstants.MAGIC_CODE_BYTES[0] != b0
                || ProtocolConstants.MAGIC_CODE_BYTES[1] != b1) {
            throw new IllegalArgumentException("Unknown magic code: " + b0 + ", " + b1);
        }
        byte version = frame.readByte();

        int fullLength = frame.readInt();
        short headLength = frame.readShort();
        byte messageType = frame.readByte();
        byte codecType = frame.readByte();
        byte compressorType = frame.readByte();
        int requestId = frame.readInt();

        RpcMessage rpcMessage = new RpcMessage();
        rpcMessage.setSerializer(codecType);
        rpcMessage.setRequestId(requestId);
        rpcMessage.setCompressor(compressorType);
        rpcMessage.setMessageType(messageType);

        //header
        int headMapLength = headLength - ProtocolConstants.V1_HEAD_LENGTH;
        if (headMapLength > 0) {
            Map<String, String> map = HeadMapSerializer.getInstance().decode(frame, headMapLength);
            rpcMessage.getHeadMap().putAll(map);
        }

        //body
        if(messageType== MsgType.HEARTBEAT_REQUEST.getCode()){
            rpcMessage.setBody(HeartbeatInvocation.PING);
        }else if (messageType == MsgType.HEARTBEAT_RESPONSE.getCode()) {
            rpcMessage.setBody(HeartbeatInvocation.PONG);
        } else {
            int bodyLength = fullLength - headLength;
            if(bodyLength>0){
                byte[] bs = new byte[bodyLength];
                frame.readBytes(bs);

                Compressor compressor = CompressorFactory.getCompressor(compressorType);
                bs = compressor.decompress(bs);
                Serializer serializer = SerializerFactory.getSerializer(codecType);
                rpcMessage.setBody(serializer.deserialize(bs));
            }
        }
        return rpcMessage;
    }
}
